//package com.xjscrm.console.chat.mq.consumer;
//
//import com.alibaba.fastjson.JSON;
//import com.xjscrm.console.chat.cache.CacheHolder;
//import com.xjscrm.console.chat.frame.send.SResultMsg;
//import com.xjscrm.console.chat.mq.dto.StaffOfflineDTO;
//import com.xjscrm.console.chat.utils.ChannelUtils;
//import io.netty.channel.Channel;
//import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
//import lombok.extern.slf4j.Slf4j;
//import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
//import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
//import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
//import org.apache.rocketmq.common.message.MessageExt;
//import org.slf4j.MDC;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.stereotype.Component;
//
//import java.util.List;
//import java.util.Objects;
//
///**
// * @author mingyang.chen create in 2024/3/12 14:10
// */
//@Slf4j
//@Component
//public class OfflineConsumer implements MessageListenerConcurrently {
//
//
//    @Autowired
//    private CacheHolder cacheHolder;
//
//    @Override
//    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
//        for (MessageExt msg : msgs) {
//            String msgBody = new String(msg.getBody());
//            StaffOfflineDTO message = JSON.parseObject(msgBody, StaffOfflineDTO.class);
//
//            try {
//                MDC.put("traceId", message.getTraceId());
//                onMessage(message);
//            } catch (Exception e) {
//                log.error("出现员工下线消息失败", e);
//                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
//            } finally {
//                MDC.clear();
//            }
//        }
//        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
//    }
//
//    private void onMessage(StaffOfflineDTO message) {
//        log.info("收到员工下线消息: {}", JSON.toJSONString(message));
//
//        Long merId = message.getMerId();
//
//        Long staffId = message.getStaffId();
//        Channel staffChannel = cacheHolder.getStaffChannel(staffId);
//        if (Objects.nonNull(staffChannel)) {
//            staffChannel.write(new TextWebSocketFrame(JSON.toJSONString(SResultMsg.fail(null, "当前员工已在其他地方登录, 即将关闭链接"))));
//            ChannelUtils.setCloseAttr(staffChannel, "员工已在其他节点上线, 当前节点关闭");
//            staffChannel.close();
//
//            cacheHolder.delStaffSessionCount(merId, staffId);
//            cacheHolder.delStaffChannel(staffId);
//        }
//    }
//}
